package com.xzx.flink.tableapi;

import com.xzx.flink.bean.ClickEvent;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * 开窗追加查询
 *
 * @version 1.0
 * @auther xinzhixuan
 * @date 2022/5/28 22:42
 */
public class TableAPI_04_AppendQueryExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 读取数据源
        SingleOutputStreamOperator<ClickEvent> eventStream = env
                .fromElements(
                        new ClickEvent("Alice", "./home", 1000L),
                        new ClickEvent("Bob", "./cart", 1000L),
                        new ClickEvent("Alice", "./prod?id=1", 25 * 60 * 1000L),
                        new ClickEvent("Alice", "./prod?id=4", 55 * 60 * 1000L),
                        new ClickEvent("Bob", "./prod?id=5", 3600 * 1000L + 60 * 1000L),
                        new ClickEvent("Cary", "./home", 3600 * 1000L + 30 * 60 * 1000L),
                        new ClickEvent("Cary", "./prod?id=7", 3600 * 1000L + 59 * 60 * 1000L)
                )
                .assignTimestampsAndWatermarks(WatermarkStrategy.<ClickEvent>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<ClickEvent>() {
                            @Override
                            public long extractTimestamp(ClickEvent element, long recordTimestamp) {
                                return element.getTimestamp();
                            }
                        }));
        // 获取表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //将 timestamp 指定为事件时间，并命名为 ts
        Table eventTable = tableEnv.fromDataStream(eventStream, $("user"), $("url"), $("timestamp").rowtime().as("ts"));

        // 将数据流转换成表
        tableEnv.createTemporaryView("event_table", eventTable);
        // 设置 1 小时滚动窗口，执行 SQL 统计查询
        Table result = tableEnv.sqlQuery("select user, window_end AS endT, COUNT(url) AS cnt " +
                "from table(tumble(table event_table, descriptor(ts), interval '1' hour)) " +// 1 小时滚动窗口
                "group by user, window_start, window_end");
        tableEnv.toDataStream(result).print("alice visit");

        env.execute();
    }
}
